package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimAsyncFunction;
import com.atguigu.gmall.realtime.beans.TradeSkuOrderBean;
import com.atguigu.gmall.realtime.utils.DateFormatUtil;
import com.atguigu.gmall.realtime.utils.MyClickhouseUtil;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.math.BigDecimal;
import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author Felix
 * @date 2023/2/8
 * SKU粒度下单业务过程聚合统计
 *  维度：sku
 *  度量：原始金额、活动减免金额、优惠券减免金额和订单金额
 * 需要启动的进程
 *      zk、kafka、maxwell、hdfs、hbase、redis、clickhouse、DwdTradeOrderDetail、DwsTradeSkuOrderWindow
 * 开发流程
 *      基本环境准备
 *      检查点相关的设置
 *      从kafka的下单明细主题中读取下单数据
 *      过滤掉null消息以及将流中的数据有jsonStr->jsonObj
 *      去重
 *          为什么会产生重复数据？
 *              在dwd层准备下单事实表数据的时候，是将订单明细、订单、订单明细活动、订单明细优惠券以及字典表这5张表
 *              关联在一起的，并且和订单明细活动、订单明细优惠券关联的时候，使用的是左外连接，如果左表数据先来
 *              会产生如下数据：
 *                  左表  null    +I
 *                  左表  null    -D
 *                  左表  右表     +I
 *              如果将上面产生数据发送到kafka主题的话，kafka主题将接收到如下消息
 *                  左表  null
 *                  null
 *                  左表  右表
 *              所以我们需要过滤掉null消息，并对重复数据进行去重
 *          去重实现方式
 *              按照唯一键进行分组
 *              状态 + 定时器|||状态 + 抵消思路
 *          状态 + 抵消思路：当第一条数据进来的时候，直接将数据发送到下游，将放到状态中；当第二条数据进来的时候，
 *              先判断状态是否为空，不为空，将状态中存储数据影响到度量值的字段进行取反，向下游传递；再将第二条数据
 *              传递到下游。
 *      将流中的数据类型由jsonObj转换为要统计的实体类对象(维度+度量)
 *      指定Watermark以及提取事件时间字段
 *      按照维度sku_id进行分组
 *      开窗
 *      聚合计算
 *      维度关联
 *          基本维度关联的实现
 *              PhoenixUtil:   List<T> queryList(String sql,Class clz)
 *              DimUtil:       JSONObject getDimInfoNoCache(String tableName,Tuple2...params)
 *          优化1：旁路缓存
 *              思路：先从缓存中获取维度数据，如果缓存中存在要关联的维度，直接将维度数据作为方法的返回值进行返回，
 *                   这种情况叫缓存命中；如果在缓存中没有找到要关联的维度，再发送请求到Phoenix表中将维度数据查询
 *                   出来，并将查询的结果放到缓存中缓存起来，方便下次查询使用
 *              选型：
 *                  状态：性能好  维护性差
 *                  Redis：性能也不错 维护性好 √
 *              关于Redis的配置
 *                  类型：String
 *                  key： dim：维度表名:主键值1_主键值2
 *                  ex: 1day
 *                  注意：一旦维度数据发生了变化，需要将缓存的数据清除掉
 *          优化2:异步IO
 *              为什么要使用异步IO？
 *                  要想提升对流中数据的处理能力，可以提升并行度，但是并行度的增加意味着需要更多的硬件资源，不可能
 *                  无限制的提升，所以我们在并行度确定的情况下，进行优化。
 *                  如果在处理流中数据的时候，使用的是map算子，在单个并行度上的处理方式是同步的，也就是说，处理完
 *                  第一个元素，再对下一个元素进行处理，这样等待的时间过长，直接影响了Flink的实时性。
 *                  所以使用异步处理的方式
 *              异步处理的语法
 *                  AsyncDataStream.[un]orderedWait(
 *                      流,
 *                      如何对流中数据进行处理 implements AsyncFunction==>asyncInvoke,
 *                      超时时间,
 *                      时间单位
 *                  )
 *              我们这里将异步处理的过程进行了封装
 *                  class DimAsyncFunction extends RichAsyncFunction{
 *                      open:获取线程池对象
 *                      asyncInvoke:
 *                          开启线程
 *                          根据流中的对象获取要关联的维度主键（抽象）
 *                          根据维度的主键获取对应的维度的对象
 *                          将维度对象相关的属性补充到流中的对象上（抽象）
 *                          将关联的结果向下游传递
 *                  }
 *              模板方法设计模式
 *                  在父类中定义了完成某一个功能的核心算法的骨架(维度关联的步骤),有些具体的实现延迟到子类中去完成
 *                  在不改变父类核心算法骨架的前提下，每一个子类都可以有自己不同的实现。
 *
 */
public class DwsTradeSkuOrderWindow {
    public static void main(String[] args) throws Exception {
        //TODO 1.基本环境准备
        //1.1 指定流处理环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2 设置并行度
        env.setParallelism(4);
        //TODO 2.检查点相关的设置(略)
        //TODO 3.从kafka主题中读取数据
        //3.1 声明消费的主题以及消费者组
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_trade_sku_order_group";
        //3.2 创建消费者对象
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        //3.3 消费数据  封装为流
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);

        //TODO 4.过滤掉null消息并对读取的数据进行类型转换   jsonStr->jsonObj
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.process(
            new ProcessFunction<String, JSONObject>() {
                @Override
                public void processElement(String jsonStr, Context ctx, Collector<JSONObject> out) throws Exception {
                    if (jsonStr != null) {
                        JSONObject jsonObj = JSON.parseObject(jsonStr);
                        out.collect(jsonObj);
                    }
                }
            }
        );
        // {"create_time":"2023-01-29 16:50:19","sku_num":"3","activity_rule_id":"4",
        // "split_original_amount":"24591.0000","sku_id":"10","source_type":"2402","date_id":"2023-01-29",
        // "source_type_name":"商品推广","user_id":"86","province_id":"34","activity_id":"2",
        // "sku_name":"Apple i","id":"732","source_id":"71","order_id":"323","split_activity_amount":"500.0",
        // "split_total_amount":"24091.0","ts":"1675846219"}
        // jsonObjDS.print(">>>>");
        //TODO 5.去重
        //5.1 按照唯一键(订单明细id)进行分组
        KeyedStream<JSONObject, String> orderDetailIdkeyedDS
            = jsonObjDS.keyBy(jsonObj -> jsonObj.getString("id"));
        /*//5.2 去重 方式1：使用状态 + 定时器
        orderDetailIdkeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> lastValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    lastValueState
                        = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("lastValueState",JSONObject.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject lastValue = lastValueState.value();
                    if(lastValue == null){
                        lastValueState.update(jsonObj);
                        long currentProcessingTime = ctx.timerService().currentProcessingTime();
                        ctx.timerService().registerProcessingTimeTimer(currentProcessingTime + 5000L);
                    }else{
                        //不精准代码：如果要用这种方式，上游dwd应该传递更精准的时间过来
                        Long curTime = jsonObj.getLong("上游聚合时间");
                        Long lastTime = lastValue.getLong("上游聚合时间");
                        if(curTime.compareTo(lastTime) > 0){
                            lastValueState.update(jsonObj);
                        }
                    }
                }

                @Override
                public void onTimer(long timestamp, OnTimerContext ctx, Collector<JSONObject> out) throws Exception {
                    JSONObject value = lastValueState.value();
                    out.collect(value);
                    lastValueState.clear();
                }
            }
        );*/
        //5.2 去重 方式2：度量指标抵销
        SingleOutputStreamOperator<JSONObject> processDS = orderDetailIdkeyedDS.process(
            new KeyedProcessFunction<String, JSONObject, JSONObject>() {
                private ValueState<JSONObject> latValueState;

                @Override
                public void open(Configuration parameters) throws Exception {
                    latValueState
                        = getRuntimeContext().getState(new ValueStateDescriptor<JSONObject>("latValueState",JSONObject.class));
                }

                @Override
                public void processElement(JSONObject jsonObj, Context ctx, Collector<JSONObject> out) throws Exception {
                    //从状态中获取上条记录
                    JSONObject lastValue = latValueState.value();
                    if(lastValue != null){
                        //说明存在重复数据  需要对度量的指标 进行抵销
                        String splitOriginalAmount = lastValue.getString("split_original_amount");
                        String splitActivityAmount = lastValue.getString("split_activity_amount");
                        String splitCouponAmount = lastValue.getString("split_coupon_amount");
                        String splitTotalAmount = lastValue.getString("split_total_amount");
                        lastValue.put("split_original_amount","-" + splitOriginalAmount);
                        lastValue.put("split_activity_amount","-" + (StringUtils.isEmpty(splitActivityAmount)?"0.0":splitActivityAmount));
                        lastValue.put("split_coupon_amount","-" + (StringUtils.isEmpty(splitCouponAmount)?"0.0":splitCouponAmount));
                        lastValue.put("split_total_amount","-" + splitTotalAmount);
                        out.collect(lastValue);
                    }
                    out.collect(jsonObj);
                    latValueState.update(jsonObj);
                }
            }
        );

        //TODO 6.对流中数据类型进行转换    jsonObj->实体类对象
        SingleOutputStreamOperator<TradeSkuOrderBean> orderBeanDS = processDS.map(
            new MapFunction<JSONObject, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(JSONObject jsonObj) throws Exception {
                    // {"create_time":"2023-01-29 16:50:19","sku_num":"3","activity_rule_id":"4",
                    // "split_original_amount":"24591.0000","sku_id":"10","source_type":"2402","date_id":"2023-01-29",
                    // "source_type_name":"商品推广","user_id":"86","province_id":"34","activity_id":"2",
                    // "sku_name":"Apple i","id":"732","source_id":"71","order_id":"323","split_activity_amount":"500.0",
                    // "split_total_amount":"24091.0","ts":"1675846219"}
                    String skuId = jsonObj.getString("sku_id");
                    String splitOriginalAmount = jsonObj.getString("split_original_amount");
                    String splitActivityAmount = jsonObj.getString("split_activity_amount");
                    String splitCouponAmount = jsonObj.getString("split_coupon_amount");
                    String splitTotalAmount = jsonObj.getString("split_total_amount");
                    Long ts = jsonObj.getLong("ts") * 1000;
                    TradeSkuOrderBean orderBean = TradeSkuOrderBean.builder()
                        .skuId(skuId)
                        .originalAmount(new BigDecimal(splitOriginalAmount))
                        .activityAmount(new BigDecimal(StringUtils.isEmpty(splitActivityAmount) ? "0.0" : splitActivityAmount))
                        .couponAmount(new BigDecimal(StringUtils.isEmpty(splitCouponAmount) ? "0.0" : splitCouponAmount))
                        .orderAmount(new BigDecimal(splitTotalAmount))
                        .ts(ts)
                        .build();
                    return orderBean;
                }
            }
        );
        // orderBeanDS.print(">>>>");

        //TODO 7.指定Watermark以及提取事件时间字段
        SingleOutputStreamOperator<TradeSkuOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
            WatermarkStrategy.<TradeSkuOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                .withTimestampAssigner(
                    new SerializableTimestampAssigner<TradeSkuOrderBean>() {
                        @Override
                        public long extractTimestamp(TradeSkuOrderBean orderBean, long recordTimestamp) {
                            return orderBean.getTs();
                        }
                    }
                )
        );
        //TODO 8.按照统计维度sku_id进行分组
        KeyedStream<TradeSkuOrderBean, String> keyedDS = withWatermarkDS.keyBy(TradeSkuOrderBean::getSkuId);

        //TODO 9.开窗
        WindowedStream<TradeSkuOrderBean, String, TimeWindow> windowDS =
            keyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));

        //TODO 10.聚合计算
        SingleOutputStreamOperator<TradeSkuOrderBean> reduceDS = windowDS.reduce(
            new ReduceFunction<TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean reduce(TradeSkuOrderBean value1, TradeSkuOrderBean value2) throws Exception {
                    value1.setOriginalAmount(value1.getOriginalAmount().add(value2.getOriginalAmount()));
                    value1.setActivityAmount(value1.getActivityAmount().add(value2.getActivityAmount()));
                    value1.setCouponAmount(value1.getCouponAmount().add(value2.getCouponAmount()));
                    value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                    return value1;
                }
            },
            new WindowFunction<TradeSkuOrderBean, TradeSkuOrderBean, String, TimeWindow>() {
                @Override
                public void apply(String groupId, TimeWindow window, Iterable<TradeSkuOrderBean> input, Collector<TradeSkuOrderBean> out) throws Exception {
                    for (TradeSkuOrderBean orderBean : input) {
                        orderBean.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                        orderBean.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                        orderBean.setTs(System.currentTimeMillis());
                        out.collect(orderBean);
                    }
                }
            }
        );
        // TradeSkuOrderBean(stt=2023-02-10 10:49:30, edt=2023-02-10 10:49:40, trademarkId=null, trademarkName=null, category1Id=null, category1Name=null, category2Id=null, category2Name=null, category3Id=null, category3Name=null, skuId=18, skuName=null, spuId=null, spuName=null, originalAmount=27597.0000, activityAmount=0.0, couponAmount=0.0, orderAmount=27597.0, ts=1675997445199)
        // reduceDS.print(">>>>>");
        //TODO 11.关联sku维度表
        /*SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = reduceDS.map(
            new MapFunction<TradeSkuOrderBean, TradeSkuOrderBean>() {
                @Override
                public TradeSkuOrderBean map(TradeSkuOrderBean orderBean) throws Exception {
                    //根据流中的对象获取要关联的维度的主键
                    String skuId = orderBean.getSkuId();
                    //根据维度的主键到维度表中获取要关联的维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo("dim_sku_info", skuId);
                    //将获取的维度对象的维度属性补充到流中的对象上
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                    return orderBean;
                }
            }
        );
        withSkuInfoDS.print(">>>");*/
        //将异步I/O操作应用于DataStream作为DataStream的一次转换操作
        SingleOutputStreamOperator<TradeSkuOrderBean> withSkuInfoDS = AsyncDataStream.unorderedWait(
            reduceDS,
            //对流中数据进行的转换操作 实现分发请求的 AsyncFunction
            new DimAsyncFunction<TradeSkuOrderBean>("dim_sku_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setSkuName(dimInfoJsonObj.getString("SKU_NAME"));
                    orderBean.setTrademarkId(dimInfoJsonObj.getString("TM_ID"));
                    orderBean.setSpuId(dimInfoJsonObj.getString("SPU_ID"));
                    orderBean.setCategory3Id(dimInfoJsonObj.getString("CATEGORY3_ID"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSkuId();
                }
            }, 60, TimeUnit.SECONDS
        );
        // withSkuInfoDS.print(">>>>");
        //TODO 12.关联spu维度表
        SingleOutputStreamOperator<TradeSkuOrderBean> withSpuInfoDS = AsyncDataStream.unorderedWait(
            withSkuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_spu_info") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setSpuName(dimInfoJsonObj.getString("SPU_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getSpuId();
                }
            }
            , 60, TimeUnit.SECONDS
        );
        // withSpuInfoDS.print(">>>");
        //TODO 13.关联tm维度表
        SingleOutputStreamOperator<TradeSkuOrderBean> withTmInfoDS = AsyncDataStream.unorderedWait(
            withSpuInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_trademark") {
                @Override
                public void join(JSONObject dimInfoJsonObj, TradeSkuOrderBean orderBean) {
                    orderBean.setTrademarkName(dimInfoJsonObj.getString("TM_NAME"));
                }

                @Override
                public String getKey(TradeSkuOrderBean orderBean) {
                    return orderBean.getTrademarkId();
                }
            }, 60, TimeUnit.SECONDS
        );
        // withTmInfoDS.print(">>>>");
        //TODO 14.关联category3维度表
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory3Stream = AsyncDataStream.unorderedWait(
            withTmInfoDS,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category3".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean)  {
                    javaBean.setCategory3Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory2Id(jsonObj.getString("category2_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory3Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 15.关联category2维度表
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory2Stream = AsyncDataStream.unorderedWait(
            withCategory3Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category2".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean)  {
                    javaBean.setCategory2Name(jsonObj.getString("name".toUpperCase()));
                    javaBean.setCategory1Id(jsonObj.getString("category1_id".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory2Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );
        //TODO 16.关联category1维度表
        SingleOutputStreamOperator<TradeSkuOrderBean> withCategory1Stream = AsyncDataStream.unorderedWait(
            withCategory2Stream,
            new DimAsyncFunction<TradeSkuOrderBean>("dim_base_category1".toUpperCase()) {
                @Override
                public void join(JSONObject jsonObj,TradeSkuOrderBean javaBean)  {
                    javaBean.setCategory1Name(jsonObj.getString("name".toUpperCase()));
                }

                @Override
                public String getKey(TradeSkuOrderBean javaBean) {
                    return javaBean.getCategory1Id();
                }
            },
            5 * 60, TimeUnit.SECONDS
        );

        //TODO 17.将关联的结果写到Clickhouse
        withCategory1Stream.print(">>>");
        withCategory1Stream.addSink(MyClickhouseUtil.getSinkFunction(
           "insert into dws_trade_sku_order_window values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"
        ));
        env.execute();
    }
}
